package com.mjf.day1

// 导入一些隐式转换，implicit
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time

object WordCountFromSocket {
  def main(args: Array[String]): Unit = {

    // 获取运行时环境，类似SparkContext
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 设置分区（又叫并行任务）的数量为1，会向资源管理器申请一个任务插槽`task slot`
    env.setParallelism(1)

    // 建立数据源
    val stream: DataStream[String] = env.socketTextStream("hadoop103", 9999, '\n')

    // 对流进行转换处理
    val transformed: DataStream[(String, Int)] = stream
      // 使用空格切分输入的字符串
      .flatMap(_.split("\\s"))
      // 类似MR的map,转换成kv形式
      .map((_, 1))
      // 按第一个字段进行聚合
      .keyBy(0)
      // 设置为5S的滚动窗口
      .timeWindow(Time.seconds(5))
      // 类似MR中的reduce,按第二个字段进行聚合
      .sum(1)

    // 将计算的结果输出到标准输出
    transformed.print()

    // 执行计算逻辑
    env.execute("WordCountFromSocket")

  }
}
